package de.distMLP;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPJob;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.TextInputFormat;
import org.apache.hama.bsp.TextOutputFormat;

import de.distMLP.messages.HamaMessage;

public class JobHelper {

	public static BSPJob createBSPJob(final HamaConfiguration conf,
			final Class<? extends BSP<LongWritable, Text, NullWritable, Text, HamaMessage>> bspClass, final String inputPath,
			final String outputPath, final int nbTasks) throws IOException {

		final BSPJob bsp = new BSPJob(conf, Train_MultilayerPerceptron.class);

		bsp.setInputFormat(TextInputFormat.class);

		bsp.setInputPath(new Path(inputPath));

		bsp.setBspClass(bspClass);

		bsp.setInputKeyClass(LongWritable.class);
		bsp.setInputValueClass(Text.class);

		bsp.setOutputKeyClass(NullWritable.class);
		bsp.setOutputValueClass(Text.class);
		bsp.setOutputFormat(TextOutputFormat.class);

		if (outputPath != null) {
			FileOutputFormat.setOutputPath(bsp, new Path(outputPath));
		}

		bsp.setNumBspTask(nbTasks);

		return bsp;
	}

	public static BSPJob createBSPJobWithoutInput(final HamaConfiguration conf,
			final Class<? extends BSP<NullWritable, NullWritable, NullWritable, Text, HamaMessage>> bspClass, final String outputPath,
			final int nbTasks) throws IOException {

		final BSPJob bsp = new BSPJob(conf, Train_MultilayerPerceptron.class);

		bsp.setInputFormat(TextInputFormat.class);

		bsp.setBspClass(bspClass);

		bsp.setInputKeyClass(LongWritable.class);
		bsp.setInputValueClass(Text.class);

		bsp.setOutputKeyClass(NullWritable.class);
		bsp.setOutputValueClass(Text.class);
		bsp.setOutputFormat(TextOutputFormat.class);

		if (outputPath != null) {
			FileOutputFormat.setOutputPath(bsp, new Path(outputPath));
		}

		bsp.setNumBspTask(nbTasks);

		return bsp;
	}
}
